package com.kingoe.canaldj.single;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * 单机Java客户端 消费 canal 消息
 * @author jinwu
 */
//@Component
//public class CanalClient implements InitializingBean {
//
//        private final static int BATCH_SIZE = 1000;
//
//        @Override
//        public void afterPropertiesSet() throws Exception {
//            // 创建链接
//            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
//            try {
//                //打开连接
//                connector.connect();
//                //订阅数据库表,全部表
//                connector.subscribe(".*\\..*");
//                //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
//                connector.rollback();
//                while (true) {
//                    // 获取指定数量的数据
//                    Message message = connector.getWithoutAck(BATCH_SIZE);
//                    //获取批量ID
//                    long batchId = message.getId();
//                    //获取批量的数量
//                    int size = message.getEntries().size();
//                    //如果没有数据
//                    if (batchId == -1 || size == 0) {
//                        try {
//                            //线程休眠2秒
//                            Thread.sleep(2000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
//                    } else {
//                        //如果有数据,处理数据
//                        printEntry(message.getEntries());
//                    }
//                    //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
//                    connector.ack(batchId);
//                }
//            } catch (Exception e) {
//                e.printStackTrace();
//            } finally {
//                connector.disconnect();
//            }
//        }
//
//        /**
//         * 打印canal server解析binlog获得的实体类信息
//         */
//        private static void printEntry(List<CanalEntry.Entry> entrys) {
//            for (CanalEntry.Entry entry : entrys) {
//                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
//                    //开启/关闭事务的实体类型，跳过
//                    continue;
//                }
//                //RowChange对象，包含了一行数据变化的所有特征
//                //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
//                CanalEntry.RowChange rowChange;
//                try {
//                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//                } catch (Exception e) {
//                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
//                }
//                //获取操作类型：insert/update/delete类型
//                CanalEntry.EventType eventType = rowChange.getEventType();
//                //打印Header信息
//                System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
//                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
//                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
//                        eventType));
//                //判断是否是DDL语句
//                if (rowChange.getIsDdl()) {
//                    System.out.println("================》;isDdl: true,sql:" + rowChange.getSql());
//                }
//                //获取RowChange对象里的每一行数据，打印出来
//                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
//                    //如果是删除语句
//                    if (eventType == CanalEntry.EventType.DELETE) {
//                        printColumn(rowData.getBeforeColumnsList());
//                        //如果是新增语句
//                    } else if (eventType == CanalEntry.EventType.INSERT) {
//                        printColumn(rowData.getAfterColumnsList());
//                        //如果是更新的语句
//                    } else {
//                        //变更前的数据
//                        System.out.println("------->; before");
//                        printColumn(rowData.getBeforeColumnsList());
//                        //变更后的数据
//                        System.out.println("------->; after");
//                        printColumn(rowData.getAfterColumnsList());
//                    }
//                }
//            }
//        }
//
//    /**
//     * 仅仅是打印出来，实际肯定会处理
//     * @param columns
//     */
//    private static void printColumn(List<CanalEntry.Column> columns) {
//            for (CanalEntry.Column column : columns) {
//                System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//        }
//    }
//
//}
